package org.tao.lightningmq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.dto.HeartBeatDTO;
import org.tao.lightningmq.common.dto.ServiceRegistryRespDTO;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.enums.ReplicationMsgTypeEnum;
import org.tao.lightningmq.nameserver.event.model.HeartBeatEvent;
import org.tao.lightningmq.nameserver.event.model.ReplicationMsgEvent;
import org.tao.lightningmq.nameserver.store.ServiceInstance;

import java.util.UUID;

/**
 * @Author lht
 * @date 2025/1/8 - 14:41
 * @description:
 */
public class HeartBeatListener implements Listener<HeartBeatEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(HeartBeatListener.class);

    @Override
    public void onReceive(HeartBeatEvent event) throws Exception {
        ChannelHandlerContext context = event.getContext();
        Object reqId = context.attr(AttributeKey.valueOf("reqId")).get();
        if (reqId == null) {
            ServiceRegistryRespDTO respDTO = new ServiceRegistryRespDTO();
            respDTO.setMsgId(event.getMsgId());
            context.writeAndFlush(new TcpMsg(
                    NameServerResponseCode.ERROR_USER_OR_PASSWORD, JSON.toJSONBytes(respDTO)));
            context.close();
            throw new IllegalAccessException("error account to connected!");
        }
//        LOGGER.info("Received heart beat data: {}", JSON.toJSONString(event));

        // 发送消息给到客户端
        if (CommonCache.getReplicationMode() == null) {
            HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
            heartBeatDTO.setMsgId(event.getMsgId());
            context.writeAndFlush(new TcpMsg(NameServerResponseCode.HEART_BEAT_SUCCESS,
                    JSON.toJSONBytes(heartBeatDTO)));
        }

        // 记录心跳时间
        String reqIdStr = (String) reqId;
        String[] reqStrArr = reqIdStr.split(":");
        ServiceInstance instance = new ServiceInstance();
        instance.setIp(reqStrArr[0]);
        instance.setPort(Integer.valueOf(reqStrArr[1]));
        instance.setLastHeartBeatTime(System.currentTimeMillis());
        CommonCache.getServiceInstanceManager().putIfExist(instance);

        // 将信息同步给从节点
        ReplicationMsgEvent replicationMsgEvent = new ReplicationMsgEvent();
        replicationMsgEvent.setServiceInstance(instance);
        replicationMsgEvent.setMsgId(event.getMsgId());
        replicationMsgEvent.setContext(event.getContext());
        replicationMsgEvent.setType(ReplicationMsgTypeEnum.HEART_BEAT.getCode());
        CommonCache.getReplicationMsgQueueManager().put(replicationMsgEvent);
    }
}
